package felix.rule.service.actors;

import felix.actor.core.TActorCtx;
import felix.actor.core.TActorException;
import felix.rule.data.plugin.ComponentLifecycleMsg;
import felix.rule.service.actors.sharde.ComponentMsgProcessor;
import lombok.extern.slf4j.Slf4j;

@Slf4j
public abstract class ComponentActor<P extends ComponentMsgProcessor> extends ContextAwareActor {

    protected final String id;
    protected final String ruleId;
    protected P processor;
    private long lastPersistedErrorTs = 0L;
    private long messagesProcessed;
    private long errorsOccurred;

    public ComponentActor(ActorSystemContext systemContext, String ruleId, String id) {
        super(systemContext);
        this.id = id;
        this.ruleId = ruleId;
    }

    abstract protected P createProcessor(TActorCtx ctx);

    @Override
    public void init(TActorCtx ctx) throws TActorException {
        super.init(ctx);
        this.processor = createProcessor(ctx);
        initProcessor(ctx);
    }

    protected void initProcessor(TActorCtx ctx) throws TActorException {
        try {
            log.debug("[{}][{}] Starting processor.", ruleId, id);
            processor.start(ctx);
//            logLifecycleEvent(ComponentLifecycleEvent.STARTED);
//            if (systemContext.isStatisticsEnabled()) {
//                scheduleStatsPersistTick();
//            }
        } catch (Exception e) {
            log.debug("[{}][{}] Failed to start processor.", ruleId, id, e);
            logAndPersist("OnStart", e, true);
//            logLifecycleEvent(ComponentLifecycleEvent.STARTED, e);
            throw new TActorException("Failed to init actor", e);
        }
    }

//    private void scheduleStatsPersistTick() {
//        try {
//            processor.scheduleStatsPersistTick(ctx, systemContext.getStatisticsPersistFrequency());
//        } catch (Exception e) {
//            log.error("[{}][{}] Failed to schedule statistics store message. No statistics is going to be stored: {}", tenantId, id, e.getMessage());
//            logAndPersist("onScheduleStatsPersistMsg", e);
//        }
//    }

    @Override
    public void destroy() {
        try {
            log.debug("[{}][{}] Stopping processor.", ruleId, id);
            if (processor != null) {
                processor.stop(ctx);
            }
//            logLifecycleEvent(ComponentLifecycleEvent.STOPPED);
        } catch (Exception e) {
            log.warn("[{}][{}] Failed to stop processor: {}", ruleId, id, e.getMessage());
            logAndPersist("OnStop", e, true);
//            logLifecycleEvent(ComponentLifecycleEvent.STOPPED, e);
        }
    }

    protected void onComponentLifecycleMsg(ComponentLifecycleMsg msg) {
        log.debug("[{}][{}][{}] onComponentLifecycleMsg: [{}]", ruleId, id, msg.getType(), msg.getEvent());
        try {
            switch (msg.getEvent()) {
                case CREATED:
                    processor.onCreated(ctx);
                    break;
                case UPDATED:
                    processor.onUpdate(ctx);
                    break;
                case ACTIVATED:
                    processor.onActivate(ctx);
                    break;
                case SUSPENDED:
                    processor.onSuspend(ctx);
                    break;
                case DELETED:
                    processor.onStop(ctx);
                    ctx.stop(ctx.getSelf());
                    break;
                default:
                    break;
            }
//            logLifecycleEvent(msg.getEvent());
        } catch (Exception e) {
            logAndPersist("onLifecycleMsg", e, true);
//            logLifecycleEvent(msg.getEvent(), e);
        }
    }

//    protected void onClusterEventMsg(PartitionChangeMsg msg) {
//        try {
//            processor.onPartitionChangeMsg(msg);
//        } catch (Exception e) {
//            logAndPersist("onClusterEventMsg", e);
//        }
//    }

//    protected void onStatsPersistTick(EntityId entityId) {
//        try {
//            systemContext.getStatsActor().tell(new StatsPersistMsg(messagesProcessed, errorsOccurred, tenantId, entityId));
//            resetStatsCounters();
//        } catch (Exception e) {
//            logAndPersist("onStatsPersistTick", e);
//        }
//    }

//    private void resetStatsCounters() {
//        messagesProcessed = 0;
//        errorsOccurred = 0;
//    }

    protected void increaseMessagesProcessedCount() {
        messagesProcessed++;
    }

    protected void logAndPersist(String method, Exception e) {
        logAndPersist(method, e, false);
    }

    private void logAndPersist(String method, Exception e, boolean critical) {
        errorsOccurred++;
        String componentName = processor != null ? processor.getComponentName() : "Unknown";
        if (critical) {
            log.debug("[{}][{}][{}] Failed to process method: {}", id, ruleId, componentName, method);
            log.debug("Critical Error: ", e);
        } else {
            log.trace("[{}][{}][{}] Failed to process method: {}", id, ruleId, componentName, method);
            log.trace("Debug Error: ", e);
        }
        long ts = System.currentTimeMillis();
        if (ts - lastPersistedErrorTs > getErrorPersistFrequency()) {
//            systemContext.persistError(tenantId, id, method, e);
            lastPersistedErrorTs = ts;
        }
    }

//    private void logLifecycleEvent(ComponentLifecycleEvent event) {
//        logLifecycleEvent(event, null);
//    }
//
//    private void logLifecycleEvent(ComponentLifecycleEvent event, Exception e) {
//        systemContext.persistLifecycleEvent(tenantId, id, event, e);
//    }

    protected abstract long getErrorPersistFrequency();
}
